/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2022-2022. All rights reserved.
 */

package com.huaweicloud.reqmt.source;

import com.huaweicloud.reqmt.domain.dwd_haydn_case_uuid_vaild_dura_stage_d;

import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Properties;

/**
 * 连接DWS数据库 建立与dwd_haydn_case_uuid_vaild_dura_stage_d表的连接
 *
 * @author l30028736
 * @since 2022-10-29
 */
public class DwsSourceDwd001 extends RichParallelSourceFunction<dwd_haydn_case_uuid_vaild_dura_stage_d> {
    private static final Logger logger = LoggerFactory.getLogger(DwsSourceDwd001.class);

    private static PreparedStatement ps;

    private static Connection conn;

    /**
     * 类在初始化的时候创建数据库连接
     */
    public DwsSourceDwd001(String username, String password, String url) {
        createConnection(username, password, url);
    }

    /**
     * 建立与DWS的连接
     */
    private void createConnection(String username, String password, String url) {
        Properties props = new Properties();
        props.setProperty("user", username);
        props.setProperty("password", password);
        props.setProperty("ssl", "false");
        try {
            Class.forName("org.postgresql.Driver");
            conn = DriverManager.getConnection(url, props);
            logger.info("Build connection success.");
            String sql =
                "SELECT ds, workspace_id, workspace_name, solution_uuid, solu_name, solu_type, test_uuid, test_name, stage, cstage, definition, start_time, end_time, dura_second FROM dwd.dwd_haydn_case_uuid_valid_dura_stage_d";
            ps = conn.prepareStatement(sql);
            logger.info("prepareStatement create success.");
        } catch (ClassNotFoundException | SQLException e) {
            logger.error("Build connection failed");
        }
    }

    /**
     * 执行SQL并发回去
     *
     * @param ctx context
     * @throws SQLException SQL执行异常
     */
    @Override
    public void run(SourceContext<dwd_haydn_case_uuid_vaild_dura_stage_d> ctx) throws SQLException {
        ResultSet query = ps.executeQuery();
        while (query.next()) {
            dwd_haydn_case_uuid_vaild_dura_stage_d data =
                new dwd_haydn_case_uuid_vaild_dura_stage_d(query.getString("ds"), query.getString("workspace_id"),
                    query.getString("workspace_name"), query.getString("solution_uuid"), query.getString("solu_name"),
                    query.getString("solu_type"), query.getString("test_uuid"), query.getString("test_name"),
                    query.getString("stage"), query.getString("cstage"), query.getString("definition"),
                    query.getString("start_time"), query.getString("end_time"), query.getString("dura_second"));
            // 发送数据
            ctx.collect(data);
        }
    }

    /**
     * 程序关闭时会释放链接
     */
    @Override
    public void cancel() {
        if (conn != null) {
            try {
                conn.close();
            } catch (SQLException e) {
                logger.error("Close connect failed: ");
            }
        }
        if (ps != null) {
            try {
                ps.close();
            } catch (SQLException e) {
                logger.error("Close prepareStatement failed");
            }
        }
    }
}
